package Demo1

import Source.SourceTest.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala.{StreamTableEnvironment, tableConversions}

object TbTest1 {

  // datastream - Table

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 读取数据
    //    val inputStream: DataStream[String] = env.socketTextStream("localhost", 7777) "E:\\Flinklearn\\Flink\\resources\\data1\\sensor.txt"
    val inputPath: String = "file:///usr/local/src/flink/sensor.txt"
    val inputDataStream: DataStream[String] = env.readTextFile(inputPath).setParallelism(1)

    // 转换样例类
    val ds: DataStream[SensorReading] = inputDataStream.map(
      data => {
        val arr: Array[String] = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )

    // 首先创建表执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // 基于流创建一张表
    val tb: Table = tableEnv.fromDataStream(ds)

    // 1.调用table api进行转换
    val tb2: Table = tb.select("id,temperature").filter("id == 'sensor_1'")
    // 2.直接用sql实现
    tableEnv.createTemporaryView("tb", tb)
    val table: Table = tableEnv.sqlQuery("select id,temperature from tb where id = 'sensor_1'")


    // 转换成追加流 进行输出
    tb2.toAppendStream[(String, Double)].print("result")
    table.toAppendStream[(String, Double)].print("result sql")


    env.execute("table api test")
  }

}
